use std::marker::PhantomData;
use super::*;

pub trait RayonBuilder<I, O1, O2, E> 
   where
        I:Send + Sync + 'static,
        O1:Send + Sync + 'static,
        O2:Send + Sync + 'static,
        E:ImmutExecutor<O1, O2>, 
        <Self::BuildReceiver as IntoIterator>::Item:Into<O1>,
{
    type Build;
    type BuildSender;
    type BuildReceiver:IntoIterator + Send + 'static;
    fn pipelined(self, exe:E) -> Self::Build;
    fn build(self) -> (Self::BuildSender, Self::BuildReceiver);
}

pub struct Pipeline<I, O, E1, E2>
    where
        I:Send + Sync + 'static,
        O:Send + Sync + 'static,
        E2:RayonPipeline<O>, 
{
    _unused_type:PhantomData<(I, O)>,
    executor1:E1,
    executor2:E2,
}

impl<I, O, E1, E2> RayonPipeline<I> for Pipeline<I, O, RayonImmutWrapper<I, O, E1>, E2>
    where 
        I:Send + Sync + 'static,
        O:Send + Sync + 'static,
        E1:ImmutExecutor<I, O>, 
        E2:RayonPipeline<O>, 
{
    type Error = E1::Error;
    fn exec<'scope>(
        &'scope self, 
        scope:&rayon::Scope<'scope>,
        input:I,
    ) -> Result<(), Self::Error> {
        self.executor1.exec(scope, input, &self.executor2)
    }
}

impl<I, O, E1, E2> Pipeline<I, O, RayonMutWrapper<I, O, E1>, E2>
    where 
        I:Send + Sync + 'static,
        O:Send + Sync + 'static,
        E1:MutExecutor<I, O>, 
        E2:RayonPipeline<O>, 
{
    pub fn exec<It>(
        self,
        input:It,
    ) -> Result<(), E1::Error> 
    where 
        It:IntoIterator + Send + 'static,
        It::Item:Into<I>,
    {
        let exe2 = &self.executor2;
        rayon::scope(move |scope|{
            let mut exe1 = self.executor1;
            for input in input {
                exe1.exec(scope, input.into(), exe2).expect("exec error");
            }
        });
        Ok(())
    }
}


impl<I, O, E> RayonBuilder<I, I, O, E> for NoneExecutor
    where
        I:Send + Sync + 'static,
        O:Send + Sync + 'static,
        E:ImmutExecutor<I, O>, 
{
    type Build = Pipeline<I, O, RayonImmutWrapper<I, O, E>, NoneExecutor>;
    type BuildSender = SendExecutor<I>;
    type BuildReceiver = ReceiverWrapper<I>;
    fn pipelined(self, exe:E) -> Self::Build {
        Pipeline {
            _unused_type:PhantomData,
            executor1:exe.into(),
            executor2:NoneExecutor,
        }
    }

    fn build(self) -> (Self::BuildSender, Self::BuildReceiver) {
        SendExecutor::channel()
    }
}

impl<I, O1, O2, O3, E1, E2, E3> RayonBuilder<I, O2, O3, E3> for Pipeline<I, O1, RayonImmutWrapper<I, O1, E1>, E2>
    where
        I:Send + Sync + 'static,
        O1:Send + Sync + 'static,
        O2:Send + Sync + 'static,
        O3:Send + Sync + 'static,
        E1:ImmutExecutor<I, O1>, 
        E2:RayonPipeline<O1> + RayonBuilder<O1, O2, O3, E3>, 
        E2::Build:RayonPipeline<O1>,
        E2::BuildSender:RayonPipeline<O1>,
        E3:ImmutExecutor<O2, O3>,
{
    type Build = Pipeline<I, O1, RayonImmutWrapper<I, O1, E1>, E2::Build>;
    type BuildSender = Pipeline<I, O1, RayonImmutWrapper<I, O1, E1>, E2::BuildSender>;
    type BuildReceiver = E2::BuildReceiver;
    fn pipelined(self, exe:E3) -> Self::Build {
        Pipeline {
            _unused_type:PhantomData,
            executor1:self.executor1.into(),
            executor2:self.executor2.pipelined(exe),
        }
    }

    fn build(self) -> (Self::BuildSender, Self::BuildReceiver) {
        let (rx, tx) = self.executor2.build();
        let rx = Pipeline {
            _unused_type:PhantomData,
            executor1:self.executor1,
            executor2:rx,
        };
        (rx, tx)
    }
}

impl<I, O1, O2, O3, E1, E2, E3> RayonBuilder<I, O2, O3, E3> for Pipeline<I, O1, RayonMutWrapper<I, O1, E1>, E2>
    where
        I:Send + Sync + 'static,
        O1:Send + Sync + 'static,
        O2:Send + Sync + 'static,
        O3:Send + Sync + 'static,
        E1:MutExecutor<I, O1>, 
        E2:RayonPipeline<O1> + RayonBuilder<O1, O2, O3, E3>, 
        E2::Build:RayonPipeline<O1>,
        E2::BuildSender:RayonPipeline<O1>,
        E3:ImmutExecutor<O2, O3>,
{
    type Build = Pipeline<I, O1, RayonMutWrapper<I, O1, E1>, E2::Build>;
    type BuildSender = Pipeline<I, O1, RayonMutWrapper<I, O1, E1>, E2::BuildSender>;
    type BuildReceiver = E2::BuildReceiver;
    fn pipelined(self, exe:E3) -> Self::Build {
        Pipeline {
            _unused_type:PhantomData,
            executor1:self.executor1.into(),
            executor2:self.executor2.pipelined(exe),
        }
    }

    fn build(self) -> (Self::BuildSender, Self::BuildReceiver) {
        let (rx, tx) = self.executor2.build();
        let rx = Pipeline {
            _unused_type:PhantomData,
            executor1:self.executor1,
            executor2:rx,
        };
        (rx, tx)
    }
}

pub struct PipelineBuilder<'ceil, 'floor, I, T, O, E1, E2> 
    where
        I:Send + Sync + 'static,
        O:Send + Sync + 'static,
        T:IntoIterator,
        T::Item:Into<I>,
        E2:RayonPipeline<O>,
{
    _unused:PhantomData<I>,
    scope:&'floor rayon::Scope<'ceil>,
    input:T,
    executor:Pipeline<I, O, E1, E2>,
}

pub fn from_scope_and_iter<'ceil, 'floor, I, It>(scope:&'floor rayon::Scope<'ceil>, iter:It) -> PipelineBuilder<'ceil, 'floor, I, It, I, (), NoneExecutor> 
    where
        I:Send + Sync + 'static,
        It:IntoIterator,
        It::Item:Into<I>,
{
    let pipeline = 
        Pipeline { _unused_type:PhantomData, executor1:(), executor2:NoneExecutor };
    PipelineBuilder { _unused: PhantomData, scope, input: iter, executor:pipeline  }
}

impl<'ceil, 'floor, I, It> PipelineBuilder<'ceil, 'floor, I, It, I, (), NoneExecutor> 
    where
        I:Send + Sync + 'static,
        It:IntoIterator,
        It::Item:Into<I>,
{
    pub fn pipelined_mut<O, E>(self, exe:E) 
        -> PipelineBuilder<'ceil, 'floor, I, It, O, RayonMutWrapper<I, O, E>, NoneExecutor>  
        where 
            O:Send + Sync + 'static,
            E:MutExecutor<I, O>,
    {
        let pipeline = 
            Pipeline { _unused_type:PhantomData, executor1:exe.into(), executor2:NoneExecutor, };
        PipelineBuilder { _unused: PhantomData, scope:self.scope, input: self.input, executor:pipeline  }
    }
    
    pub fn pipelined_immut<O, E:ImmutExecutor<I, O>>(self, exe:E) 
        -> PipelineBuilder<'ceil, 'floor, I, It, O, RayonImmutWrapper<I, O, E>, NoneExecutor>  
        where 
            O:Send + Sync + 'static,
    {
        let pipeline = 
            Pipeline { _unused_type:PhantomData, executor1:exe.into(), executor2:NoneExecutor, };
        PipelineBuilder { _unused: PhantomData, scope:self.scope, input: self.input, executor:pipeline  }
    }
}

impl<'ceil, 'floor, I, T, O, E1, E2> PipelineBuilder<'ceil, 'floor, I, T, O, RayonImmutWrapper<I, O, E1>, E2> 
    where
        I:Send + Sync + 'static,
        O:Send + Sync + 'static,
        T:IntoIterator + Send + 'static,
        T::Item:Into<I>,
        E1:ImmutExecutor<I, O>,
        E2:RayonPipeline<O>,
{
    pub fn pipelined_mut<I2, O2, E3>(
        self, 
        executor:E3
    ) -> PipelineBuilder<'ceil, 'floor, I2, E2::BuildReceiver, O2, RayonMutWrapper<I2, O2, E3>, NoneExecutor> 
    where 
        I2:Send + Sync + 'static,
        O2:Send + Sync + 'static,
        E2:RayonBuilder<O, I2, O2, RayonImmutPhantom>,
        E3:MutExecutor<I2, O2>,
        E2::Build:RayonPipeline<O>,
        E2::BuildSender:RayonPipeline<O>,
    {
        let (run_executor, next_iter) = self.executor.build();
        self.scope.spawn(move |_|{
            let run_executor = run_executor;
            let run_executor_ref = &run_executor;
            rayon::scope(move |scope| {
                for input in self.input {
                    run_executor_ref.exec(scope, input.into()).expect("exec error");
                }
            })
        });
        let new_executor = from_scope_and_iter(self.scope, next_iter).pipelined_mut(executor);
        new_executor
    }

    pub fn pipelined_immut<I2, O2, E3>(
        self, 
        executor:E3
    ) -> PipelineBuilder<'ceil, 'floor, I, T, O, RayonImmutWrapper<I, O, E1>, E2::Build> 
    where 
        I2:Send + Sync + 'static,
        O2:Send + Sync + 'static,
        E2:RayonBuilder<O, I2, O2, E3>,
        E3:ImmutExecutor<I2, O2>,
        E2::Build:RayonPipeline<O>,
    {
        let pipeline = Pipeline {
            _unused_type:PhantomData, 
            executor1: self.executor.executor1, 
            executor2: self.executor.executor2.pipelined(executor),
        };
        PipelineBuilder { _unused: PhantomData, scope:self.scope, input: self.input, executor: pipeline }
    }

    pub fn run(self) {
        rayon::spawn(move ||{
            let run_executor = self.executor;
            let run_executor_ref = &run_executor;
            rayon::scope(move |scope| {
                for input in self.input {
                    run_executor_ref.exec(scope, input.into()).expect("exec error");
                }
            })
        });
    }
}

impl<'ceil, 'floor, I, T, O, E1, E2> PipelineBuilder<'ceil, 'floor, I, T, O, RayonMutWrapper<I, O, E1>, E2> 
    where
        I:Send + Sync + 'static,
        O:Send + Sync + 'static,
        T:IntoIterator + Send + 'static,
        T::Item:Into<I>,
        E1:MutExecutor<I, O>,
        E2:RayonPipeline<O>,
{
    pub fn pipelined_mut<I2, O2, E3>(
        self, 
        executor:E3
    ) -> PipelineBuilder<'ceil, 'floor, I2, E2::BuildReceiver, O2, RayonMutWrapper<I2, O2, E3>, NoneExecutor> 
    where 
        I2:Send + Sync + 'static,
        O2:Send + Sync + 'static,
        E2:RayonBuilder<O, I2, O2, RayonImmutPhantom>,
        E3:MutExecutor<I2, O2>,
        E2::Build:RayonPipeline<O>,
        E2::BuildSender:RayonPipeline<O>,
    {
        let (run_executor, next_iter) = self.executor.build();
        self.scope.spawn(move |_|{run_executor.exec(self.input).expect("exec error"); });
        let new_executor = from_scope_and_iter(self.scope, next_iter).pipelined_mut(executor);
        new_executor
    }

    pub fn pipelined_immut<I2, O2, E3>(
        self, 
        executor:E3
    ) -> PipelineBuilder<'ceil, 'floor, I, T, O, RayonMutWrapper<I, O, E1>, E2::Build> 
    where 
        I2:Send + Sync + 'static,
        O2:Send + Sync + 'static,
        E2:RayonBuilder<O, I2, O2, E3>,
        E3:ImmutExecutor<I2, O2>,
        E2::Build:RayonPipeline<O>,
    {
        let pipeline = Pipeline {
            _unused_type:PhantomData, 
            executor1: self.executor.executor1, 
            executor2: self.executor.executor2.pipelined(executor),
        };
        PipelineBuilder { _unused: PhantomData, scope:self.scope, input: self.input, executor: pipeline }
    }

    pub fn run(self) {
        rayon::spawn(move ||{self.executor.exec(self.input).expect("exec error"); });
    }
}